
#include <stdio.h>
#include <sys/types.h>
#include <time.h>
#include <string.h>
#include <stdlib.h>
#include <sstream>
#include <iomanip>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/filesystem.hpp>
using boost::system::error_code;
using boost::system::system_category;
using boost::system::system_error;

#include "bocat/common/base64.hpp"

#include "fc_indexiter.hpp"
#include "filecell.hpp"


namespace	sakcab	{

#define	IFSP_CACHE_SIZE		32

const std::string   FileCell::OP_PUT( "P" );
const std::string   FileCell::OP_GET( "G" );
const std::string   FileCell::OP_DEL( "D" );

const std::string	FileCell::SLICE_BASENAME( "slice" );
const std::string	FileCell::DELETE_FILENAME( "DELETED" );

const u_int32_t     FileCell::MIN_SLICE_SIZE_THRESHOLD = 1024 * 1024 * 32;
const u_int32_t     FileCell::MAX_SLICE_SIZE_THRESHOLD = 1024 * 1024 * 1024;


///////////////////////////////////////////////////////////////////////
//

FileCell::FileCell( const std::string data_dir, u_int32_t file_rotate_size, int slim )
		:	_data_dir( data_dir ),
			_file_rotate_size( file_rotate_size ),
			_slim( slim ),
			_ofs_append(),
			_ofs_delete(),
			_ifsp_cache( IFSP_CACHE_SIZE ),
			_index_iterator_p( NULL ),
			_buf_p(NULL),
			_buf_size( 0 )
{
	if ( 0 == _file_rotate_size ) {
		_file_rotate_size = MAX_SLICE_SIZE_THRESHOLD;
	}
	if ( _file_rotate_size < MIN_SLICE_SIZE_THRESHOLD ) {
		_file_rotate_size = MIN_SLICE_SIZE_THRESHOLD;
	}
	if ( _file_rotate_size > MAX_SLICE_SIZE_THRESHOLD ) {
		_file_rotate_size = MAX_SLICE_SIZE_THRESHOLD;
	}

	_slim_lowater = int( _slim * 0.9 );
	if ( _slim_lowater < 1 ) {
		_slim_lowater = 1;
	}
	_slimming = false;

	initialize();

	RecordLocation	dummy_loc;
	_dummy_loc_b64 = bocat::common::base64_encode( (const char *)(&dummy_loc), sizeof(dummy_loc) );
}

FileCell::~FileCell()
{
	finalize();
}

///////////////////////////////////////////////////////////////////////
//

void
FileCell::initialize()
{
	_min_fileno = 0;
	_max_fileno = 0;

	scanDataDirectory();

}

void
FileCell::finalize()
{
	if (_ofs_append.is_open()) 
		_ofs_append.close();

	if (_ofs_delete.is_open())
		_ofs_delete.close();

	closeAllReadFiles();

	if (_buf_p) {
		free( _buf_p );

		_buf_p = NULL;
		_buf_size = 0;
	}

	if (_index_iterator_p) {
		delete _index_iterator_p;
		_index_iterator_p = NULL;
	}
}

///////////////////////////////////////////////////////////////////////
//


int 
FileCell::put( const std::string& skey_b64,
		 const std::string& kpin_b64,
		 const std::string& key,
		 const std::string& data,
		 Timestamp&			ts,
		 RecordLocation&    loc,
		 const RecordLocation& prev_loc )
{
	std::ofstream&	ofs = openAppendFile();
	
	std::string	prev_loc_b64 = bocat::common::base64_encode( (const char *)(&prev_loc), sizeof(prev_loc) );

	struct timeval	tv = ts.timeval();
	std::ostringstream	line;

	// 
	// Field-ID     0       1      2     3           4         5          6    7
	// Field-Name tv_sec  tv_usec  op  skey_b64  kpin_b64  prev_loc_b64  key  data 
	line << tv.tv_sec << "\t" << tv.tv_usec << "\t" << OP_PUT << "\t" << skey_b64 << "\t" << kpin_b64 << "\t" << prev_loc_b64 << "\t" << key << "\t" << data << std::endl;

	loc.fileno = _max_fileno;
	loc.offset = ofs.tellp();
	loc.size = line.str().size();	 

	ofs << line.str();

	if (ofs.tellp() > _file_rotate_size ) {
		ofs.close();
	}

	//fprintf(stderr, "%s:%d PUT() key:%s loc:%s\n", __FILE__, __LINE__, key.c_str(), loc.str().c_str() );
	return	0;
}

int 
FileCell::get( const std::string& skey_b64,
		 const std::string& kpin_b64,
		 const std::string& key,
			   std::string& data,
		 Timestamp&			ts,
		 const RecordLocation&  loc,
		       RecordLocation& prev_loc )
{ 
	//fprintf(stderr, "%s:%d GET() key:%s loc:%s\n", __FILE__, __LINE__, key.c_str(), loc.str().c_str() );
	std::ifstream&	ifs = openReadFile( loc.fileno );

	ifs.seekg( loc.offset, std::ios::beg );
	u_int32_t	where = ifs.tellg();
	if ( where != loc.offset ) {
		return -1;
	}

	/*
	char*	buf_p = getBuf( loc.size );
	*(buf_p + loc.size - 1) = 0;
	ifs.getline( buf_p, loc.size );
	u_int32_t	data_got_size = ifs.gcount();
	if ( (loc.size - data_got_size) > 1 ) {
		return -2;
	}

	char*	p_tab = buf_p;
	// skip skey_b64
	p_tab = strchr( p_tab + 1, '\t' );
	if ( NULL == p_tab ) {
		return -3;
	}
	// skip kpin_b64
	p_tab = strchr( p_tab + 1, '\t' );
	if ( NULL == p_tab ) {
		return -3;
	}
	// skip key
	p_tab = strchr( p_tab + 1, '\t' );
	if ( NULL == p_tab ) {
		return -3;
	}

	++p_tab;
	data.assign( p_tab, data_got_size - (p_tab - buf_p) );
	*/

	std::string	line;
	std::getline( ifs, line );

	// 
	// Field-ID     0       1      2     3           4         5          6    7
	// Field-Name tv_sec  tv_usec  op  skey_b64  kpin_b64  prev_loc_b64  key  data 
	std::vector< std::string >	fields;
	boost::split(fields, line, boost::is_any_of("\t"));
	if ( fields.size() != 8 ) {
		return	-5;
	}
	const std::string&	tv_sec_s = fields[0];
	const std::string&	tv_usec_s = fields[1];

	struct	timeval	tv;
	tv.tv_sec = boost::lexical_cast< time_t >( tv_sec_s.c_str() );
	tv.tv_usec = boost::lexical_cast< suseconds_t >( tv_usec_s.c_str() );
	ts = Timestamp( tv );

	const std::string&	prev_loc_b64 = fields[5];
	std::string	_prev_loc = bocat::common::base64_decode< std::string >( prev_loc_b64 );
	if ( _prev_loc.size() != sizeof(prev_loc) ) {
		return	-6;
	}
	memcpy( &prev_loc, _prev_loc.c_str(), sizeof(prev_loc) );

	data = fields[ 7 ];

	return	0;
}

static	const std::string	ERASE_DATA( "D" );

int 
FileCell::erase( const std::string&   skey_b64,
		   const std::string&   kpin_b64,
		   const std::string&   key,
		   const Timestamp&		ts,
		   const RecordLocation&    loc )
{
	//
	//std::ofstream&	ofs = openDeleteFile();
	//
	//ofs << skey_b64 << "\t" << kpin_b64 << "\t" << key << "\t"
	//	<< loc.fileno << "\t" << loc.offset << "\t" << loc.size << std::endl;

	std::ofstream&	ofs = openAppendFile();
	
	struct timeval	tv = ts.timeval();
	std::ostringstream	line;

	// 
	// Field-ID     0       1      2     3           4         5          6    7
	// Field-Name tv_sec  tv_usec  op  skey_b64  kpin_b64  prev_loc_b64  key  data 
	line << tv.tv_sec << "\t" << tv.tv_usec << "\t" << OP_DEL << "\t" << skey_b64 << "\t" << kpin_b64 << "\t" << _dummy_loc_b64 << "\t" << key << "\t" << ERASE_DATA << std::endl;

	ofs << line.str();

	if (ofs.tellp() > _file_rotate_size ) {
		ofs.close();
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//



int
FileCell::put( std::vector< Record >& record_v, Timestamp& ts )
{
	std::ofstream&	ofs = openAppendFile();
	
	struct timeval	tv = ts.timeval();
	for (size_t k=0; k < record_v.size(); ++k ) {
		Record&	r = record_v[ k ];

		std::string	prev_loc_b64;
		prev_loc_b64 = bocat::common::base64_encode( (const char *)(&r.prev_loc), sizeof(r.prev_loc) );

		// 
		// Field-ID     0       1      2     3           4         5          6    7
		// Field-Name tv_sec  tv_usec  op  skey_b64  kpin_b64  prev_loc_b64  key  data 
		std::ostringstream	line;
		line << tv.tv_sec << "\t" << tv.tv_usec << "\t" << OP_PUT << r.skey_b64 << "\t" << r.kpin_b64 << "\t" << prev_loc_b64 << "\t" << r.key << "\t" << r.data << std::endl;

		r.loc.fileno = _max_fileno;
		r.loc.offset = ofs.tellp();
		r.loc.size = line.str().size();	 

		ofs << line.str();
	}

	if (ofs.tellp() > _file_rotate_size ) {
		ofs.close();
	}

	return	0;
}

int
FileCell::append( std::vector< Record >& record_v, Timestamp& ts )
{
	std::ofstream&	ofs = openAppendFile();
	
	struct timeval	tv = ts.timeval();
	for (size_t k=0; k < record_v.size(); ++k ) {
		Record&	r = record_v[ k ];

		std::string	prev_loc_b64;
		if ( k > 0 ) {
			const Record& r_p = record_v[ k - 1 ];
			if ( r.skey_b64 == r_p.skey_b64 ) {
				prev_loc_b64 = bocat::common::base64_encode( (const char *)(&r_p.loc), sizeof(r_p.loc) );
			}
			else {
				prev_loc_b64 = bocat::common::base64_encode( (const char *)(&r.prev_loc), sizeof(r.prev_loc) );
			}
		}
		else {
			prev_loc_b64 = bocat::common::base64_encode( (const char *)(&r.prev_loc), sizeof(r.prev_loc) );
		}

		std::ostringstream	line;
		line << tv.tv_sec << "\t" << tv.tv_usec << "\t" << OP_PUT << r.skey_b64 << "\t" << r.kpin_b64 << "\t" << prev_loc_b64 << "\t" << r.key << "\t" << r.data << std::endl;

		r.loc.fileno = _max_fileno;
		r.loc.offset = ofs.tellp();
		r.loc.size = line.str().size();	 

		ofs << line.str();
	}

	if (ofs.tellp() > _file_rotate_size ) {
		ofs.close();
	}

	return	0;
}


int
FileCell::erase( std::vector< Record >& record_v, Timestamp& ts )
{
	//std::ofstream&	ofs = openDeleteFile();
	//	
	//for (size_t k=0; k < record_v.size(); ++k ) {
	//	Record&	r = record_v[ k ];
	//
	//	ofs << r.skey_b64 << "\t" << r.kpin_b64 << "\t" << r.key << "\t"
	//		<< r.loc.fileno << "\t" << r.loc.offset << "\t" << r.loc.size << std::endl;
	//}

	std::ofstream&	ofs = openAppendFile();
	struct timeval	tv = ts.timeval();

	for (size_t k=0; k < record_v.size(); ++k ) {
		Record&	r = record_v[ k ];

		// 
		// Field-ID     0       1      2     3           4         5          6    7
		// Field-Name tv_sec  tv_usec  op  skey_b64  kpin_b64  prev_loc_b64  key  data 
		std::ostringstream	line;
		line << tv.tv_sec << "\t" << tv.tv_usec << "\t" << OP_DEL << "\t" << r.skey_b64 << "\t" << r.kpin_b64 << "\t" << _dummy_loc_b64 << "\t" << r.key << "\t" << ERASE_DATA << std::endl;

		ofs << line.str();
	}

	if (ofs.tellp() > _file_rotate_size ) {
		ofs.close();
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//

std::ofstream&
FileCell::openAppendFile( void )
{
	if (_ofs_append.is_open()) {
		return _ofs_append;
	}

	// find the only APPEND file
	std::string	fn_append = getAppendFileName();
	_ofs_append.open( fn_append.c_str(), std::ios::app );
	if ( !_ofs_append ) {
		std::ostringstream	msg;
		msg << "fail to open append file " << fn_append;
		throw std::runtime_error( msg.str() );
	}

	return _ofs_append;
}

std::ofstream&
FileCell::openDeleteFile( void )
{
	if (_ofs_delete.is_open()) {
		return _ofs_delete;
	}

	// open the only DELETE file
	std::string	fn_delete = getDeleteFileName();
	_ofs_delete.open( fn_delete.c_str(), std::ios::app );
	if ( !_ofs_delete ) {
		std::ostringstream	msg;
		msg << "fail to open delete log file " << fn_delete;
		throw std::runtime_error( msg.str() );
	}

	return _ofs_delete;
}


std::string
FileCell::getAppendFileName( void )
{
	boost::filesystem::path	data_path( _data_dir );
	
	if (_max_fileno > 0) {
		std::string	fn = getCurrentAppendFileName();

		boost::filesystem::path	append_file = data_path / fn;
		if (boost::filesystem::exists( append_file ) ) {
			if (boost::filesystem::file_size(append_file) < _file_rotate_size) {
				return append_file.string();
			}
		}
	}

	std::string	fn = getNextAppendFileName();
	boost::filesystem::path	append_file = data_path / fn;
	return append_file.string();
}


std::string
FileCell::getDeleteFileName( void )
{
	boost::filesystem::path	data_path( _data_dir );

	boost::filesystem::path	delete_path = data_path / DELETE_FILENAME;

	return delete_path.string();
}


std::string
FileCell::getNextAppendFileName( void )
{
	++_max_fileno;

	return getFileName( _max_fileno );
}

std::string
FileCell::getCurrentAppendFileName( void )
{
	return getFileName( _max_fileno );
}


std::string
FileCell::getFileName( u_int32_t fileno )
{
	std::ostringstream	fn;

	fn << SLICE_BASENAME << "." << std::setw(9) << std::setfill('0') << fileno;

	return fn.str();
}


///////////////////////////////////////////////////////////////////////
//

std::ifstream&  
FileCell::openReadFile( u_int32_t fileno )
{
	boost::mutex::scoped_lock	_scoped_lock( this->_ifsp_cache_mutex );

	ifsp_type	ifsp;

	if ( _ifsp_cache.exists( fileno ) ) {
		ifsp = _ifsp_cache.get( fileno );
		return *ifsp;
	}

	ifsp = new std::ifstream();
	std::string	fn = getReadFileName( fileno );
	ifsp->open( fn.c_str() );
	if ( !(ifsp->good()) ) {
		std::ostringstream	msg;
		msg << "fail to open file " << fn;
		throw std::runtime_error( msg.str() );
	}

	ifsp_type	ifsp_removed = _ifsp_cache.put( fileno, ifsp );
	if ( ifsp_removed ) {
		if (ifsp_removed->is_open()) {
			ifsp_removed->close();
		}

		delete ifsp_removed;
	}

	return *ifsp;
}


void
FileCell::closeAllReadFiles( void )
{
	boost::mutex::scoped_lock	_scoped_lock( this->_ifsp_cache_mutex );

	ifsp_cache_type::Key_List	_list( _ifsp_cache.get_all_keys() );
	for( ifsp_cache_type::Key_List_Iter liter = _list.begin(); liter != _list.end(); liter++ ) {
		ifsp_type	_ifsp = _ifsp_cache.get( *liter, false );

		if ( _ifsp->is_open() ) {
			_ifsp->close();
		}
		delete _ifsp ;
	}

	_ifsp_cache.clear();
}


std::string
FileCell::getReadFileName( u_int32_t fileno )
{
	boost::filesystem::path	data_path( _data_dir );
	
	std::string	fn = getFileName( fileno );
	boost::filesystem::path	file_path = data_path / fn;
	return file_path.string();
}


///////////////////////////////////////////////////////////////////////
//


char*
FileCell::getBuf( u_int32_t buf_size )
{
	if (buf_size <= _buf_size) {
		return	_buf_p;
	}

	char*	p = (char *)realloc(_buf_p, buf_size);
	if (NULL == p) {
		std::ostringstream	msg;
		msg << "fail to malloc buffer (size:" << buf_size << ")";
		throw std::runtime_error( msg.str() );
	}

	_buf_p = p;
	_buf_size = buf_size;

	return	_buf_p;
}


///////////////////////////////////////////////////////////////////////
//

void
FileCell::scanDataDirectory( void )
{
	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directories( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				std::string	en = boost::filesystem::extension( itr->path() );
				u_int32_t	fn = 0;
				try {
					fn = boost::lexical_cast< u_int32_t >( en.c_str() + 1 );	// skip '.'
				}
				catch( std::exception& e ) {
					continue;
				}

				if ( (0 == _min_fileno) || (_min_fileno > fn) ) {
						_min_fileno = fn;
				}
				if ( (0 == _max_fileno) || (_max_fileno < fn) ) {
						_max_fileno = fn;
				}
			}
		}
	}

}


///////////////////////////////////////////////////////////////////////
//

IndexIterator*
FileCell::getSlimIndexIterator( void )
{
	if (_slim < 1) {
		return	NULL;
	}

	if (!_slimming) {
		int	count = countFileToSlim();
		if ( count < _slim ) {
			return	NULL;
		}

		_slimming = true;
	}

	boost::mutex::scoped_lock	_scoped_lock( this->_index_iterator_mutex );
	if ( NULL != _index_iterator_p ) {
		// already in sliming
		return	NULL;
	}

	std::string	file_to_slim;
	int	fileno_to_slim = getFileToSlim( file_to_slim ); 
	if ( fileno_to_slim > 0 ) {
		_file_slimming = file_to_slim;
		_fileno_slimming = fileno_to_slim;

		_index_iterator_p = new FileCellIndexIterator( file_to_slim.c_str() );
		return _index_iterator_p;
	}
	else {
		// no more file to slim
		_slimming = false;
	}

	return	NULL;
}

void
FileCell::freeSlimIndexIterator( IndexIterator* it, bool done )
{
	if ( (NULL != it) && (it == _index_iterator_p) ) {
		delete _index_iterator_p;

		if (done && (_file_slimming.size() > 0)) {
			boost::filesystem::path	slim_file( _file_slimming );
			try {
				boost::filesystem::remove( slim_file );
			}
			catch( ... ) {
				// pass
			}
		}

		_index_iterator_p = NULL;
	}
}


int
FileCell::getFileToSlim( std::string& file_to_slim )
{
	if (_slim < 1) {
		return	0;
	}

	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directory( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	int	count_files = 0;
	u_int32_t	oldest_size = 0;
	std::time_t	oldest_mtime = time(NULL) + 86400;
	boost::filesystem::path	oldest_file;

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				++count_files;

				u_int32_t	file_size = boost::filesystem::file_size( itr->path() );
				if ( file_size < _file_rotate_size ) {
					// skip the last file
					continue;
				}

				time_t	mtime = boost::filesystem::last_write_time( itr->path() );
				if (mtime < oldest_mtime) {
					oldest_mtime = mtime;
					oldest_file = itr->path();
				}
				else if (mtime == oldest_mtime) {
					if ( file_size > oldest_size ) {
						oldest_size = file_size;
						oldest_file = itr->path();
					}
				}
			}
		}
	}

	if (count_files <= _slim_lowater) {
		return	0;
	}

	file_to_slim = oldest_file.string();
	int	fileno = getFileNoFromName( file_to_slim );

	return	fileno;
}

int
FileCell::getFileNoFromName( const std::string& filename )
{
	boost::filesystem::path	path( filename );

	std::string	ext = boost::filesystem::extension( path );
	int	fileno = 0;
	try {
		// skip '.000001'
		fileno = boost::lexical_cast< int >( ext.c_str() + 1 );
	}
	catch( ... ) {
	}

	return	fileno;
}

int
FileCell::countFileToSlim( void )
{
	int	count_files = 0;

	boost::filesystem::path	data_path( _data_dir );

	if ( !boost::filesystem::exists( data_path ) ) {
		if ( !boost::filesystem::create_directory( data_path ) ) {
			throw boost::filesystem::filesystem_error( "fail to create directory", data_path, error_code(errno, system_category) );
		}
	}

	boost::filesystem::directory_iterator end_itr; // default construction yields past-the-end
	for ( boost::filesystem::directory_iterator itr( data_path );
		itr != end_itr;
		++itr ) {
		if (boost::filesystem::is_regular_file( itr->status() ) ) {
			std::string	bn = boost::filesystem::basename( itr->path() );
			if ( bn == SLICE_BASENAME ) {
				++count_files;
			}
		}
	}

	return	count_files;
}

///////////////////////////////////////////////////////////////////////
//

}	//// namespace sakcab

